package com.central.car.streaming

import java.util.Properties

import com.alibaba.fastjson.{JSON, JSONObject}
import com.central.face.streaming.{SaveEsData, TransData}
import com.central.utils._
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, Row, SaveMode, SparkSession}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.elasticsearch.spark.rdd.EsSpark

object CarStreaming {

  val CAR_URL: String = ConfigUtil.getProperties().getProperty("mysql.url")
  val TABLE_RESULT: String = ConfigUtil.getProperties().getProperty("mysql.car.table.result")
  val TABLE_DETAIL: String = ConfigUtil.getProperties().getProperty("mysql.car.table.detail")
  val SAVE_ES_INDEX: String = ConfigUtil.getProperties().getProperty("car.save.es.index")

  def main(args: Array[String]): Unit = {

    val configs = ConfigUtil.getProperties()
    val KAFKA_SERVERS = configs.getProperty("bootstrap.servers")
    val KAFKA_GROUPID = configs.getProperty("group.id")
    val OFFSET = configs.getProperty("auto.offset.reset")
    val CAR_TOPIC = configs.getProperty("car.topic")
    val mysql_username = configs.getProperty("mysql.username")
    val mysql_password = configs.getProperty("mysql.password")
    val elasticsearch_host = configs.getProperty("elasticsearch.host", "192.168.5.180")
    val elasticsearch_port = configs.getProperty("elasticsearch.port", "9200")

    val conf = new SparkConf().setAppName("CarStreaming")
      //      .setMaster("spark://192.168.5.180:7077")
      .setMaster("local[8]")
      .set("es.nodes", elasticsearch_host)
      .set("es.port", elasticsearch_port)
      .set("es.mapping.date.rich", "false")


    val sc = new SparkContext(conf)
    sc.setLogLevel("ERROR")
    val ssc = new StreamingContext(sc, Seconds(120))
    val session = SparkSession.builder().config(conf).getOrCreate()

    val properties = new Properties()
    properties.put("user", mysql_username)
    properties.put("password", mysql_password)
    val prop = sc.broadcast(properties)
    val deviceParam: Broadcast[Dataset[Row]] = ssc.sparkContext.broadcast(TransData.transFaceData(session))


    val kafkaparam = Map(
      "bootstrap.servers" -> KAFKA_SERVERS,
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> KAFKA_GROUPID,
      "auto.offset.reset" -> OFFSET,
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )
    val topics = Array(CAR_TOPIC)

    val stream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics, kafkaparam)
    )

    stream.map(x => {
      val str = x.value()
      //      deviceId + "_" + car_id + "_" + time + "_" + address + "_" + deviceName + "_" + longitude + "_" + latitude+"_"+carpath+"_"+cartype
      val nObject = JSON.parseObject(str)
      val device_id = nObject.getString("device_code")
      val car_id = nObject.getString("car_id")
      val time = nObject.getString("captime")
      val address = nObject.getString("address")
      val deviceName = nObject.getString("device_name")
      val longitude = nObject.getString("longitude")
      val latitude = nObject.getString("latitude")
      val carpath = nObject.getString("car_path")
      val cartype = nObject.getString("car_type")
      (device_id, car_id, time, address, deviceName, longitude, latitude, carpath, cartype)
      //      val strings = str.split("_")

      //      FaceData(strings(0), strings(1), strings(2), strings(3))
      //      (strings(0), strings(1), strings(2), strings(3), strings(4), strings(5), strings(6), strings(7),strings(8))
    }).filter(x => {
      (x._3.length == 13)
    })
      .foreachRDD(rdds => {
        val cacherdds = rdds.cache()
        val carRdd = cacherdds.map(x => {
          (x._1, x._2 + "\\u001" + x._3 + "\\u001" + x._4 + "\\u001" + x._5 + "\\u001" + x._6 + "\\u001" + x._7 + "\\u001" + x._8 + "\\u001" + x._9)
        })
        //        广播变量的值为(device_code,siteId)与车牌数据(device_code,详情)关联得到(siteId,详情)
        val site_car = deviceParam.value.rdd.map(x => {
          (x.getLong(0) + "", x.getLong(1) + "")
        })
          .join(carRdd)
          .map(x => {
            x._2
          }).cache()

        //根据人像数据中点位和时间信息，查询es得到侦码数据


        //广播变量的值为(siteId,device_code)与人像数据(siteId,详情)进行join后得到(device_code,time+详情)
        val deviceCode_time: RDD[(String, String)] = deviceParam.value.rdd.map(x => {
          (x.getLong(1) + "", x.getLong(0) + "")
        })
          .join(site_car)
          .map(x => {
            x._2
          }).cache()
        if (deviceCode_time.count() > 0) {

          //根据(device_code,time+详情)查询出es中的数据
          val esRddNew = LoadESData.loadDataFromES(deviceCode_time)
            .coalesce(8, false)
            .cache()
          println(esRddNew.count())
          //查询出来的es数据转换为(device_code,详情)，再与广播变量(device_code,siteId)进行关联，
          // 使每条详情与siteId关联，进而通过siteId与人像数据关联。
          val code_esRdd = esRddNew.map(x => {
            (x._1, x)
          })
          //得到(siteId,es数据详情)
          val site_esDetailRdd = deviceParam.value.rdd.map(x => {
            (x.getLong(0) + "", x.getLong(1) + "")
          })
            .join(code_esRdd)
            .map(x => {
              val site = x._2._1
              val detail = x._2._2
              (site, detail._1 + "\\u001" + detail._2 + "\\u001" + detail._3 + "\\u001" + detail._4 + "\\u001" + detail._5 + "\\u001" + detail._6 + "\\u001" + detail._7 + "\\u001" + detail._8 + "\\u001" + detail._9)
            })
            .cache()
          //          val esRdd = LoadESData.loadDataFromESOld(cacherdds).cache()
          //          val faceClueRdd: RDD[(String, String)] = esRdd.map(esrdd => {
          //
          //            //deviceId, code type captime address deviceName longitude latitude
          //            (esrdd._1, esrdd._2 + "_" + esrdd._3 + "_" + esrdd._4 + "_" + esrdd._5 + "_" + esrdd._6 + "_" + esrdd._7 + "_" + esrdd._8) //(deviceId,mac/imsi,type)
          //          })

          //通过人像数据与侦码数据进行关联，得到关联数据
          //          val result: RDD[(String, (String, Option[String]))] = praperRdd.leftOuterJoin(faceClueRdd)
          val results: RDD[(String, (String, Option[String]))] = site_car.leftOuterJoin(site_esDetailRdd)
          val joinRdd = results
            .filter(joinResult => {
              joinResult._2._2.isDefined
            })
            .map(filterResult => {
              //deviceId,  mac/imsi,  id
              //              val deviceId = filterResult._1
              val codeMessage = filterResult._2._2.get
              val CarAndTime = filterResult._2._1

              val car_message = CarAndTime.split("\\\\u001")
              val car = car_message(0)
              val car_captime = car_message(1)
              val car_address = car_message(2)
              val car_deviceName = car_message(3)
              val car_longitude = car_message(4)
              val car_latitude = car_message(5)

              val car_path = car_message(6)
              var car_type = car_message(7)

              val code_message = codeMessage.split("\\\\u001")
              val deviceId = code_message(0)
              val code = code_message(1)
              val code_types = code_message(2)
              val code_capTime = code_message(3)
              val code_address = code_message(4)
              val code_deviceName = code_message(5)
              val code_longitude = code_message(6)
              val code_latitude = code_message(7)

              val es_id = code_message(8)
              //              val code = code_message(0)
              //              val code_types = code_message(1)
              //              val code_capTime = code_message(2)
              //              val code_address = code_message(3)
              //              val code_deviceName = code_message(4)
              //              val code_longitude = code_message(5)
              //              val code_latitude = code_message(6)


              (car + "_" + code, //1
                deviceId, //2
                code_types, //3
                code_capTime, //4
                code_address,
                code_deviceName,
                code_longitude,
                code_latitude,
                car_captime,
                car_address,
                car_deviceName,
                car_longitude,
                car_latitude
                , car_path
                , es_id
                , car_type
              )
            }).sortBy(x => {
            x._4
          })

          //将关联数据存储到es
          val saveEsRdd = joinRdd.map(x => {
            //faceAndCode,face,code,time,place
            SaveEsCarData(x._1, x._2, x._3, x._4, x._5, x._6, x._7, x._8, x._9, x._10, x._11, x._12, x._13,
              //              , x._1 + x._9
              x._14,
              x._15,
              x._16
            )
          }).distinct()

          import session.implicits._
          //通过redis对关联信息进行计数
          //flag, key, count, face_captime, codeType, savedata
          val accomulatedCarRdd: RDD[(Boolean, String, String, String, String, SaveEsCarData, String)] = saveEsRdd.distinct().map(x => {
            RedisUtil.accomulatorCarByString(x.car_code, x.car_captime, x.code_types, x, x.code_capTime)
          })
            .cache()

          val saveESRdd = accomulatedCarRdd.filter(x => {
            !x._1
          })
          EsSpark.saveToEs(saveESRdd.map(x => {
            val saveData: SaveEsCarData = x._6
            val json = new JSONObject()
            val car_code = saveData.car_code.split("_")
            json.put("licenseplate", car_code(0))
            json.put("tag_code", car_code(1))
            json.put("captime", saveData.car_captime)
            json.put("dataid", saveData.id)
            json.put("licenseplatepath", saveData.car_path)
            json
          }), SAVE_ES_INDEX)

          val notSaveEsRdd = accomulatedCarRdd.filter(x => {
            x._1
          })
          val insert = saveESRdd.filter(x => {
            x._3.equals("5")
          })
          if (insert.count() > 0) {
            insert.map(x => {
              val car_code = x._6.car_code.split("_")
              InsertCarData(car_code(0), car_code(1), x._6.code_types, x._3.toInt, x._6.car_path, x._6.car_type, System.currentTimeMillis() + "", 50.00)

            })
              .toDF().write.mode(SaveMode.Append).jdbc(CAR_URL, TABLE_RESULT, prop.value)
            //======================================================================
            //            val insertRdd = LoadESData.loadCarHistory(insert)
            //            insertRdd.map(x => {
            //              (x._1, x._3 + "_" + x._4)
            //            })
            //              .groupByKey()
            //              .map(x => {
            //                val car_code = x._1.split("_")
            //                val count = x._2.size + ""
            //                val times: Iterable[String] = x._2
            //                val code_type = times.map(x => {
            //                  val type_times = x.split("_")
            //                  type_times(0)
            //                }).toList.head
            //                val time = times
            //                  .map(x => {
            //                    x.split("_")(1)
            //                  })
            //                  .toList.sortBy(x => {
            //                  x
            //                })
            //                val firstTime = time.head
            //                val lastTime = time(time.length - 1)
            //                val car = car_code(0)
            //                val code = car_code(1)
            //                InsertCarData(car, code, code_type, count.toInt, ParseTime.DateToLong(firstTime), ParseTime.DateToLong(lastTime), 50.00)
            //              })
            //              .toDF().write.mode(SaveMode.Append).jdbc(CAR_URL, TABLE_RESULT, prop.value)
            //========================================================================

            //====================================================
            //            insertRdd.map(x => {
            //
            //              SaveMysqlCarData(
            //                x._1.split("_")(0),
            //                x._1.split("_")(1),
            //                x._2.toInt,
            //                x._3,
            //                ParseTime.DateToLong(x._4),
            //                x._5, x._6, x._7, x._8, ParseTime.DateToLong(x._9), x._10, x._11, x._12, x._13, x._2.toInt
            //              )
            //            })
            //              .toDF().write.mode(SaveMode.Append).jdbc(CAR_URL, TABLE_DETAIL, prop.value)
            //=====================================================
          }
          //达到阈值的数据进行更新操作
          if (notSaveEsRdd.count() > 0) {
            val updataParamRdd = notSaveEsRdd.filter(x => {
              !x._3.equals("5")
            }).cache()

            if (updataParamRdd.count() > 0) {
              val update: Array[(String, String, String, String)] = updataParamRdd
                .map(x => {
                  val key = x._2
                  val count = x._3
                  val time = x._4
                  val code_type = x._5
                  (key, count, time, code_type)
                  //                SaveData.updateResult(key, count, time, code_type)
                })
                .groupBy(x => {

                  x._1
                })
                .map(
                  x => {
                    val head: (String, String, String, String) = x._2.toArray.sortBy(x => {
                      Integer.parseInt(x._2)
                    }).reverse.head
                    head
                  })
                .collect()
              SaveData.updateCarResult(update)

              //============================================================
              //              updataParamRdd.map(x => {
              //                x._6
              //              })
              //                .map(x => {
              //                  SaveMysqlCarData(
              //                    x.car_code.split("_")(0),
              //                    x.car_code.split("_")(1),
              //                    x.deviceId.toInt,
              //                    x.code_types, ParseTime.DateToLong(x.code_capTime), x.code_address, x.car_deviceName, x.code_longitude, x.code_latitude,
              //                    ParseTime.DateToLong(x.car_captime), x.car_address, x.car_deviceName, x.car_longitude, x.car_latitude, x.deviceId.toInt
              //                  )
              //                })
              //                .toDF().write.mode(SaveMode.Append).jdbc(CAR_URL, TABLE_DETAIL, prop.value)
              //=========================================================

            }
          }
        }
      })
    ssc.start()
    ssc.awaitTermination()
  }
}

case class SaveEsCarData(car_code: String,
                         deviceId: String,
                         code_types: String,
                         code_capTime: String,
                         code_address: String,
                         code_deviceName: String,
                         code_longitude: String,
                         code_latitude: String,
                         car_captime: String,
                         car_address: String,
                         car_deviceName: String,
                         car_longitude: String,
                         car_latitude: String,
                         car_path: String,
                         id: String,
                         car_type: String
                        )

case class SaveMysqlCarData(carid: String, //1
                            code: String, //2
                            car_deviceid: Integer, //3
                            code_type: String, //4
                            code_captime: java.lang.Long, //5
                            code_address: String, //6
                            code_device_name: String, //7
                            code_longitude: String, //8
                            code_latitude: String, //9
                            car_captime: java.lang.Long, //10
                            car_address: String, //11
                            car_device_name: String, //12
                            car_longitude: String, //13
                            car_latitude: String, //14
                            code_device_id: Integer)

case class InsertCarData(licenseplateid: String,
                         tagcode: String,
                         codetype: String,
                         counts: Integer,
                         licenseplate_path: String,
                         vehicletype: String,
                         create_time: String,
                         //                         first_time: java.lang.Long,
                         //                         last_time: java.lang.Long,
                         confidence: Double
                        )